package com.atguigu.flink01;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/2/18
 * 该案例演示了程序链式编写以及泛型问题
 */
public class Flink04_WC_Unbound_Socket_Generic {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();
        env
                .socketTextStream("hadoop102",8888)
                //使用匿名内部类创建接口实现类对象
                /*.flatMap(
                        new FlatMapFunction<String, Tuple2<String,Integer>>() {
                            @Override
                            public void flatMap(String lineStr, Collector<Tuple2<String, Integer>> out) throws Exception {
                                String[] words = lineStr.split(" ");
                                for (String word : words) {
                                    out.collect(Tuple2.of(word,1));
                                }
                            }
                        }
                )*/
                //通过lambda表达式创建接口实现类对象
                .flatMap(
                        (String lineStr, Collector<Tuple2<String, Integer>> out) -> {
                            String[] words = lineStr.split(" ");
                            for (String word : words) {
                                out.collect(Tuple2.of(word,1));
                            }
                        }
                )
                //.returns(new TypeHint<Tuple2<String, Integer>>() {})
                //注意：如果使用lambda表达式  会存在泛型擦除问题
                .returns(Types.TUPLE(Types.STRING,Types.INT))
//                .keyBy(0)
                .keyBy(tup->tup.f0)
                .sum(1)
                .print();

        env.execute();

    }

}
